/************************************************************************
 *  Copyright (c) 2011-2012 HONG LEIMING.
 *
 * Permission is hereby granted, free of charge, to any person obtaining a
 * copy of this software and associated documentation files (the
 * "Software"), to deal in the Software without restriction, including
 * without limitation the rights to use, copy, modify, merge, publish,
 * distribute, sublicense, and/or sell copies of the Software, and to permit
 * persons to whom the Software is furnished to do so, subject to the
 * following conditions:
 *
 * The above copyright notice and this permission notice shall be included
 * in all copies or substantial portions of the Software.
 * 
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
 * OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
 * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
 * NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
 * DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
 * OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
 * USE OR OTHER DEALINGS IN THE SOFTWARE.
 ***************************************************************************/
/************************************************************************
 *  Copyright (c) 2011-2012 HONG LEIMING.
 *
 * Permission is hereby granted, free of charge, to any person obtaining a
 * copy of this software and associated documentation files (the
 * "Software"), to deal in the Software without restriction, including
 * without limitation the rights to use, copy, modify, merge, publish,
 * distribute, sublicense, and/or sell copies of the Software, and to permit
 * persons to whom the Software is furnished to do so, subject to the
 * following conditions:
 *
 * The above copyright notice and this permission notice shall be included
 * in all copies or substantial portions of the Software.
 * 
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
 * OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
 * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
 * NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
 * DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
 * OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
 * USE OR OTHER DEALINGS IN THE SOFTWARE.
 ***************************************************************************/
package net.zbus;

import net.zbus.ZMQ.Context;
import net.zbus.ZMQ.Socket;

public class BusClient {

	private String id;
	private String host = "localhost";
	private int port = 15555;

	private Context ctx;
	private Socket socket;

	protected BusClient(Context ctx) {
		this(ctx, "localhost", 15555);
	}

	protected BusClient(Context ctx, String host, int port) {
		this(ctx, host, port, null);

	}

	protected BusClient(Context ctx, String host, int port, String id) {
		if (ctx == null)
			throw new IllegalArgumentException("context is null");

		this.ctx = ctx;
		this.host = host;
		this.port = port;
		this.id = id;
		this.reconnect();
	}

	public boolean reconnect() {
		if (this.socket != null) {
			this.socket.close();
		}
		this.socket = ctx.socket(ZMQ.DEALER);
		this.socket.setLinger(0);
		String address = String.format("tcp://%s:%d", host, port);
		return this.socket.connect(address);
	}

	public ZMsg request(String service, ZMsg message){
		return this.request(service, "", message, 1000);
	}
	
	public ZMsg request(String service, String token, ZMsg message){
		return this.request(service, token, message, 10000);
	}
	
	public ZMsg request(String service, ZMsg message, int timeout){
		return this.request(service, "", message, timeout);
	}
	
	public ZMsg request(String service, String token, ZMsg message, int timeout) {
		this.socket.setReceiveTimeOut(timeout);
		message.pushFront("");
		message.pushFront(token);
		message.pushFront(service);
		message.pushFront(Protocol.MDPC);
		message.pushFront("");

		message.send(this.socket);
		ZMsg res = ZMsg.recvMsg(this.socket);
		if (res != null) {
			if (res.frameSize() < 3) {
				throw new ZBusException(
						"[<empty>, <header>, <msg_id>] frames required");
			}
			res.popFront();
			res.popFront();
			res.popFront();
			return res;
		} else {
			this.reconnect();
			throw new ZBusException("request timeout");
		}
	}

	public ZMsg send(AsynCtrl ctrl, ZMsg msg) {
		this.socket.setReceiveTimeOut(ctrl.getTimeout());

		msg.pushFront(ctrl.getMessageId());
		msg.pushFront(ctrl.getPeerId());
		msg.pushFront(ctrl.getToken());
		msg.pushFront(ctrl.getService());
		msg.pushFront(Protocol.MDPQ);
		msg.pushFront("");

		msg.send(this.socket);

		ZMsg res = ZMsg.recvMsg(this.socket);
		if (res != null) {
			if (res.frameSize() < 3) {
				throw new ZBusException(
						"[<empty>, <header>, <msg_id>] frames required");
			}
			res.popFront();
			res.popFront();
			res.popFront();
			return res;
		} else {
			this.reconnect();
			throw new ZBusException("request timeout");
		}
	}

	public boolean publish(String service, String token, String topic,
			ZMsg message, int timeout) {
		AsynCtrl ctrl = new AsynCtrl();
		ctrl.setService(service);
		ctrl.setToken(token);
		ctrl.setTimeout(timeout);

		message.pushFront(topic);
		ZMsg res = this.send(ctrl, message);
		String status = res.popFrontStr();
		if (status != null && status.equals("200")) {
			return true;
		}
		return false;
	}

	public boolean broadcast(String service, String token, ZMsg message,
			int timeout) {
		AsynCtrl ctrl = new AsynCtrl();
		ctrl.setService(service);
		ctrl.setToken(token);
		ctrl.setTimeout(timeout);

		ZMsg res = this.send(ctrl, message);
		String status = res.popFrontStr();
		if (status != null && status.equals("200")) {
			return true;
		}
		return false;
	}

	public ZMsg recv(int probeInterval) {
		this.socket.setReceiveTimeOut(probeInterval);
		while (true) {
			ZMsg msg = ZMsg.recvMsg(this.socket);
			if (msg == null) {
				int rc = this.probe(2500);
				if (rc != 0) {
					this.reconnect();
				}
			} else {
				msg.popFront();
				msg.popFront(); // header
				return msg;
			}
		}
	}

	public int probe(int timeout) {
		this.socket.setReceiveTimeOut(timeout);
		ZMsg msg = new ZMsg();
		msg.pushFront(Protocol.MDPT);
		msg.pushFront("");

		msg.send(this.socket);
		ZMsg res = ZMsg.recvMsg(this.socket);
		if (res != null) {
			return 0;
		} else {
			return -1;
		}
	}

	public ZMsg monitor(String token, ZMsg message) {
		return this.monitor(token, message, 10000);
	}
	
	public ZMsg monitor(ZMsg message) {
		return this.monitor("", message, 10000);
	}
	
	public ZMsg monitor(String token, ZMsg message, int timeout) {
		this.socket.setReceiveTimeOut(timeout);
		message.pushFront(token);
		message.pushFront(Protocol.MDPM);
		message.pushFront("");

		message.send(this.socket);
		ZMsg res = ZMsg.recvMsg(this.socket);
		if (res != null) {
			if (res.frameSize() < 3) {
				throw new ZBusException(
						"[<empty>, <header>, <msg_id>] frames required");
			}
			res.popFront();
			res.popFront();
			res.popFront();
			return res;
		} else {
			this.reconnect();
			throw new ZBusException("request timeout");
		}
	}

	public void destroy() {
		if (this.socket != null) {
			this.socket.close();
			this.socket = null;
		}
	}

	public String getId() {
		return id;
	}

	public void setId(String id) {
		this.id = id;
	}

	public String getHost() {
		return host;
	}

	public void setHost(String host) {
		this.host = host;
	}

	public int getPort() {
		return port;
	}

	public void setPort(int port) {
		this.port = port;
	}

	public static void main(String[] args) {
		Context ctx = ZMQ.context(1);

		BusClient client = new BusClient(ctx);

		ZMsg req = new ZMsg();
		req.pushBack("ls");
		
		ZMsg res = client.monitor("", req, 2500); 
		while(res != null && !res.isEmpty()){
			System.out.println(res.popFrontStr());
		}

		client.destroy();

		ctx.term();
	}
}
